
#include "BondMarketStreamV2.h"

#include <core/time/date.h>
#include <uam/ServerTimeMgr.h>
#include <uam/UserAccountManager.h>

#include <QtGlobal>

#include "RealTimeDealList.h"
#include "RealTimeDealStat.h"
#include "RealTimeDealUtil.h"

typedef MarketStreamRec Stat;
typedef MarketStreamInfo Record;

#ifndef BASSERTS
#define BASSERTS  //
#endif            // !BASSERTS

using namespace detail;
namespace detail {
// compatible
bool IsExistInVec(const std::vector<uint32_t>& vec, int ncount,
                  uint32_t nvalue) {
  for (int i = 0; i < ncount && i < (int)vec.size(); i++) {
    if (vec[i] == nvalue) return true;
  }
  return false;
}
}  // namespace detail
bool CMarketStream::IsBondPassType(const CBondInfo& bond, MarktType type) {
  if (bond.IsValidBond()) {
    if (type == CMarketStream::MarktAll) return true;
    if (type == CMarketStream::MarktMunicp && bond.IsMunicipal())
      return true;
    else if (type == MSUtil::GetMarketStreamType(bond.GetBondSubType(),
                                                 bond.GetRateType())) {
      if (type == CMarketStream::MarktFinance ||
          type == CMarketStream::MarktNFRate ||
          type == CMarketStream::MarktCountry ||
          type == CMarketStream::MarktShort ||
          type == CMarketStream::MarktChina ||
          type == CMarketStream::MarktEnterpr ||
          type == CMarketStream::MarktOther)
        return true;
    }
  }
  return false;
}
bool CMarketStream::IsBondPassPage(const CBondInfo& bond,
                                   emMktStreamPage pageType) {
  bool bValid = false;
  switch (pageType) {
    case CMarketStream::mktPage_Interest:
      if (bond.IsRatingDebt()) bValid = true;
      break;
    case CMarketStream::mktPage_Credit:
      if (!bond.IsRatingDebt() && !bond.IsBondNCD()) bValid = true;
      break;
    case CMarketStream::mktPage_TFRelative:
      if (MSUtil::IsBondRelativeType(bond) && MSUtil::IsBondRelativeTerm(bond))
        bValid = true;
      break;
    case CMarketStream::mktPage_NCD:
      if (bond.IsBondNCD()) bValid = true;
      break;
    case CMarketStream::mktPage_Credit_Inc_NCD:
      if (!bond.IsRatingDebt()) bValid = true;
      break;
    default:
      break;
  }
  return bValid;
}
int CMarketStream::GetMarketInfoKey(int stat, int mktType,
                                    const CBondInfo& bond) {
  int key = -1;
  if (stat == kStatTypeByType) {
    if (mktType == CMarketStream::MarktCountry ||
        mktType == CMarketStream::MarktFinance ||
        mktType == CMarketStream::MarktShort ||
        mktType == CMarketStream::MarktChina ||
        mktType == CMarketStream::MarktEnterpr ||
        mktType == CMarketStream::MarktOther) {
      key = 1;
    }
  } else if (stat == CMarketStream::kStatTypeByDate) {
    std::string str =
        bond.GetRemainPeriod(ServerTimeMgr::instance().serverTime());
    int ldays = (int)MSUtil::CalcRemainingDay(str);

    if (ldays < 180)
      key = CMarketStream::DateRange1;

    else if (ldays >= 180 && ldays < 365)
      key = CMarketStream::DateRange2;

    else if (ldays >= 365 && ldays < 365 * 3)
      key = CMarketStream::DateRange3;

    else if (ldays >= 365 * 3 && ldays < 365 * 5)
      key = CMarketStream::DateRange4;

    else if (ldays >= 365 * 5 && ldays < 365 * 7)
      key = CMarketStream::DateRange5;

    else if (ldays >= 365 * 7 && ldays < 365 * 10)
      key = CMarketStream::DateRange6;

    else if (ldays >= 365 * 10)
      key = CMarketStream::DateRange7;
  } else if (stat == CMarketStream::kStatTypeTransDate) {
    std::string str =
        bond.GetRemainPeriod(ServerTimeMgr::instance().serverTime());
    int ldays = (int)MSUtil::CalcRemainingDay(str);
    if (mktType == CMarketStream::MarktShort) {
      if (ldays < 30)
        key = CMarketStream::DateRange1;
      else if (ldays >= 30 && ldays < 90)
        key = CMarketStream::DateRange2;
      else if (ldays >= 90 && ldays < 180)
        key = CMarketStream::DateRange3;
      else if (ldays >= 180 && ldays < 270)
        key = CMarketStream::DateRange4;
      else if (ldays >= 270 && ldays < 365)
        key = CMarketStream::DateRange5;
    } else {
      if (ldays < 180)
        key = CMarketStream::DateRange1;
      else if (ldays >= 180 && ldays < 365)
        key = CMarketStream::DateRange2;
      else if (ldays >= 365 && ldays < 365 * 3)
        key = CMarketStream::DateRange3;
      else if (ldays >= 365 * 3 && ldays < 365 * 5)
        key = CMarketStream::DateRange4;
      else if (ldays >= 365 * 5 && ldays < 365 * 7)
        key = CMarketStream::DateRange5;
      else if (ldays >= 365 * 7 && ldays < 365 * 10)
        key = CMarketStream::DateRange6;
      else if (ldays >= 365 * 10)
        key = CMarketStream::DateRange7;
    }
  } else if (stat == CMarketStream::kStatTypeRatioType) {
    if (strcmp(bond.GetBondSubType(), "SCB") == 0) {
      key = CMarketStream::MarktCenter;
    } else if (strcmp(bond.GetBondSubType(), "TLB") == 0 ||
               strcmp(bond.GetBondSubType(), "LLB") == 0) {
      key = CMarketStream::MarktLocal;
    } else if (mktType == CMarketStream::MarktCountry ||
               mktType == CMarketStream::MarktFinance ||
               mktType == CMarketStream::MarktShort ||
               mktType == CMarketStream::MarktChina ||
               mktType == CMarketStream::MarktEnterpr ||
               mktType == CMarketStream::MarktOther) {
      key = mktType;
    }
  } else if (stat == kStatTypeTransRating) {
    const char* strBondRaing = bond.GetIssuerRating();
    if (strcmp(strBondRaing, "AAA") == 0)
      key = CMarketStream::RatingAAA;
    else if (strcmp(strBondRaing, "AA+") == 0)
      key = CMarketStream::RatingAAP;
    else if (strcmp(strBondRaing, "AA") == 0)
      key = CMarketStream::RatingAA;
    else if (strcmp(strBondRaing, "AA-") == 0)
      key = CMarketStream::RatingAAN;
    else if (strcmp(strBondRaing, "A+") == 0)
      key = CMarketStream::RatingAP;
    else if (strcmp(strBondRaing, "A") == 0)
      key = CMarketStream::RatingA;
    else
      key = CMarketStream::RatingOther;
  } else if (stat == CMarketStream::kStatTypeTransNFRate) {
    const char* strBondRaing = bond.GetRateType();
    if (strcmp(strBondRaing, "SHIBOR") == 0)
      key = CMarketStream::RatypeShibor;
    else if (strcmp(strBondRaing, "DEPO") == 0)
      key = CMarketStream::RatypeDepo;
  }
  return key;
}
bool CMarketStream::IsDealValid(const xQBMarketStreamUnit_c& msu,
                                uint32_t& unDay, kStreamType& type) {
  // 用于导入成交数据时的过滤判断
  // 只有有效的符合要求的成交数据才会导入
  // 内部函数,不对外公开
  if (0 != strcmp("2", msu.m_type)) {
    log_warning("[STREAM][INVALID]非成交信息 m_type:%s %s indexID:%d",
                msu.m_type, msu.m_company_id /*,msu.m_company_name*/,
                msu.m_indexID);
    return false;
  }
  unDay =
      qb::base::IntDate::timeToInt(qMax(msu.m_create_time, msu.m_modify_time));
  if (unDay <= 0) {
    log_warning(
        "[STREAM][INVALID]非法创建时间或者更改时间 create:%64u  modify:%64u "
        "indexID:%d",
        msu.m_create_time, msu.m_modify_time, msu.m_indexID);
    return false;
  }
  if (strcmp(msu.m_company_id, "e") == 0) {
    type = CMarketStream::kStreamTypeExchange;
    return true;
  }
  if (strcmp(msu.m_company_id, "c") == 0) {
    type = type = CMarketStream::kStreamTypeCFets;
    return true;
  }
  int nBondIndex = CBondContainer::instance().GetBondIndex(
      msu.m_body2.m_bondkey, msu.m_body2.m_listedmarket);
  if (!CBondContainer::instance().IsValidIndex(nBondIndex)) {
    log_warning("[STREAM][INVALID]找不到债券 key:%s.%s indexID:%d",
                msu.m_body2.m_bondkey, msu.m_body2.m_listedmarket,
                msu.m_indexID);
    return false;
  }
  if (CBondContainer::instance().ElementAtR(nBondIndex).IsBondNCD()) {
    bool bNCDValid =
        SSUserAccountManager::instance().IsBrokerNCDAvailable(msu.m_company_id);
    if (!bNCDValid) {
      log_warning(
          "[STREAM][INVALID]NCD不可用 key:%s.%s indexID:%d companyid:%s",
          msu.m_body2.m_bondkey, msu.m_body2.m_listedmarket, msu.m_indexID,
          msu.m_company_id);
      return false;
    }
  } else {
    bool bBondValid = SSUserAccountManager::instance().IsBrokerBondAvailable(
        msu.m_company_id);
    if (!bBondValid) {
      log_warning(
          "[STREAM][INVALID]Broker债券不允许 key:%s.%s indexID:%d companyid:%s",
          msu.m_body2.m_bondkey, msu.m_body2.m_listedmarket, msu.m_indexID,
          msu.m_company_id);
      return false;
    }
  }
  type = CMarketStream::kStreamTypeBond;
  return true;
}

bool CMarketStream::FilterDealList(xQBMarketStreamList_c* pLstdata,
                                   std::vector<ParsedStreamInfo>& batch) {
  // 初步过滤,静态函数
  uint32_t unDay = 0;
  int ncount = 0;
  if (pLstdata == nullptr || pLstdata->m_List.size() <= 0) return false;
  batch.clear();
  batch.reserve(pLstdata->m_List.size());
  CMarketStreamBase::kStreamType type = CMarketStreamBase::kStreamTypeBond;
  std::list<xQBMarketStreamUnit_c>::iterator it = pLstdata->m_List.begin();
  for (; it != pLstdata->m_List.end(); ++it) {
    const xQBMarketStreamUnit_c& msu = *it;
    if (CMarketStream::IsDealValid(msu, unDay, type))  // NCD权限检查
    {
      BASSERTS(unDay > 0, "");
      ParsedStreamInfo si;
      si.unDay = unDay;
      si.type = type;
      detail::MSUtil::DecodeMarketStreamUnit(si.info, msu);
      batch.push_back(si);
      ncount++;
    }
  }
  return ncount > 0;
}

bool CMarketStream::FilterCFETSDealList(xQBMarketStreamList_c* pLstdata,
                                        std::vector<ParsedStreamInfo>& batch,
                                        uint32_t DealType) {
  // 初步过滤,静态函数
  int ncount = 0;
  if (pLstdata == nullptr || pLstdata->m_List.size() <= 0) return false;
  batch.clear();
  batch.reserve(pLstdata->m_List.size());
  CMarketStreamBase::kStreamType type = CMarketStreamBase::kStreamTypeCFets;
  char sCid[16] = {0};
  FMTBUF(sCid, "%d", DealType);
  std::list<xQBMarketStreamUnit_c>::iterator it = pLstdata->m_List.begin();
  for (; it != pLstdata->m_List.end(); ++it) {
    const xQBMarketStreamUnit_c& msu = *it;
    uint32_t unDay = qb::base::IntDate::timeToInt(
        qMax(msu.m_create_time, msu.m_modify_time));
    BASSERTS(unDay > 0, "company_id:%s id:%s", msu.m_company_id,
             msu.m_body2.m_id);
    ParsedStreamInfo si;
    si.unDay = unDay;
    si.type = type;
    detail::MSUtil::DecodeMarketStreamUnit(si.info, msu);
    FIELDCOPY(si.info.m_company_id, sCid);
    batch.push_back(si);
    ncount++;
  }
  return ncount > 0;
}

bool CMarketStream::UpdateMarketInfoMap(const MarketStreamInfo& record,
                                        int mktype, int stattype,
                                        CMarketInfoMap& infoMap) {
  // GetMarketInfoKey 用于获取计数统计的key
  // 如果检测失败,该函数返回-1,那么将不用于统计
  // typedef CMarketStream::MarktType MarktType;
  // typedef CMarketStream::kStatType kStatType;
  int nIndex = CBondContainer::instance().GetBondIndex(
      record.m_body.m_bondkey, record.m_body.m_listedmarket);
  if (!CBondContainer::instance().IsValidIndex(nIndex)) return false;

  const CBondInfo& bond = CBondContainer::instance().ElementAtR(nIndex);
  if (IsBondPassType(bond, (MarktType)mktype)) {
    kStatType stat = (kStatType)stattype;
    int key = GetMarketInfoKey(stattype, mktype, bond);
    if (key >= 0) {
      infoMap.m_map[key] += 1;
      return true;
    }
  }
  return false;
}

CMarketStream::CMarketStream()
    : m_tag(OBJECT_TAG),  // TODO 干啥的？
      m_nextid(1),
      m_last(nullptr) {}

CMarketStream::~CMarketStream() {
  SCOPED_MUTEX(m_lock);
  m_last = nullptr;
  for (DayDealMap::iterator it = m_deal_map.begin(); it != m_deal_map.end();
       ++it) {
    RTDealList* dealList = it->second;
    if (dealList) delete dealList;
  }
  m_deal_map.clear();
}

CMarketStream& CMarketStream::cfectsStream() {
  static CMarketStream s_cfets;
  return s_cfets;
}

CMarketStream& CMarketStream::exchangeStream() {
  static CMarketStream s_ex;
  return s_ex;
}

CMarketStream& CMarketStream::bondStream() {
  static CMarketStream s_bond;
  return s_bond;
}

bool CMarketStream::Clear(uint32_t uTime) {
  SCOPED_MUTEX(m_lock);
  m_last = nullptr;
  if (uTime <= 0) {
    DayDealMap::iterator it = m_deal_map.begin();
    for (; it != m_deal_map.end(); ++it) {
      RTDealList* rtlist = it->second;
      if (rtlist) delete rtlist;
    }
    m_deal_map.clear();
  } else {
    DayDealMap::iterator it = m_deal_map.find(uTime);
    if (it != m_deal_map.end()) {
      RTDealList* rtlist = it->second;
      if (rtlist) delete rtlist;
      m_deal_map.erase(it);
    }
  }

  return true;
}

int CMarketStream::GetVersion(int companyID) {
  SCOPED_MUTEX(m_lock);
  RTDealList* dealList = GetOrCreateList(GetBondDealInvalidateDay());
  if (dealList) return dealList->GetVersion(companyID);
  return 0;
}

int CMarketStream::GetRecords(IRealTimeDealModel* keys, int& pos, int count,
                              void* records) {
  SCOPED_MUTEX(m_lock);
  RTDealList* dealList = GetOrCreateList(GetBondDealInvalidateDay());
  if (dealList) return dealList->GetRecords(keys, pos, count, records);
  return 0;
}
bool CMarketStream::GetMarketStreamStatByKey(const char* szCombBondKey,
                                             MarketStreamRec& stat,
                                             uint32_t timeDay) {
  uint32_t UnDay = timeDay <= 0 ? GetBondDealInvalidateDay() : timeDay;
  SCOPED_MUTEX(m_lock);
  RTDealList* rdlist = GetList(UnDay);
  if (rdlist) {
    return rdlist->GetStat(szCombBondKey, stat);
  }
  return false;
}
bool CMarketStream::GetBondDeal(const char* szCombBondKey,
                                const char* szCompanyId,
                                MarketStreamInfo& record, time_t t,
                                uint32_t day) {
  /*
          用以替代CMarketStream中的GetMarketStreamByKey和GetNearByMarketStream接口
          时间复杂度:O(N)
          算法优化:map<szCombBondKey,list,DealIndex> 存起来;
  */
  if (szCombBondKey == nullptr || szCompanyId == nullptr) return false;
  char szBondKey[32] = {0};
  char szListedMarket[32] = {0};
  {
    CBondContainer::ScopedLock l(CBondContainer::instance());
    int nBondIndex = CBondContainer::instance().GetBondIndex(szCombBondKey);
    if (nBondIndex < 0) return false;
    const CBondInfo& bi = CBondContainer::instance().ElementAtR(nBondIndex);
    FIELDCOPY(szBondKey, bi.GetBondKey());
    FIELDCOPY(szListedMarket, bi.GetListedMarket());
  }

  uint32_t UnDay = day <= 0 ? GetBondDealInvalidateDay() : day;

  SCOPED_MUTEX(m_lock);

  RTDealList* rdlist = GetList(UnDay);
  if (rdlist == nullptr) return false;

  return rdlist->FindRecord(szBondKey, szListedMarket, szCompanyId, record, t);
}

bool CMarketStream::GetBondDealBatch(const std::vector<BrokerKey>& brokerKeys,
                                     std::vector<MarketStreamInfo>& records,
                                     time_t t, uint32_t day) {
  uint32_t UnDay = day <= 0 ? GetBondDealInvalidateDay() : day;
  records.reserve(brokerKeys.size());
  SCOPED_MUTEX(m_lock);
  RTDealList* rdlist = GetList(UnDay);
  if (rdlist) {
    for (int i = 0; i < (int)brokerKeys.size(); ++i) {
      MarketStreamInfo msi;
      const BrokerKey& key = brokerKeys[i];
      char szBondKey[32] = {0};
      char szListedMarket[32] = {0};
      const CBondInfo& bi = CBondContainer::instance().ElementAtR(key.m_nIndex);
      FIELDCOPY(szBondKey, bi.GetBondKey());
      FIELDCOPY(szListedMarket, bi.GetListedMarket());
      rdlist->FindRecord(szBondKey, szListedMarket, key.GetBrokerId().c_str(),
                         msi, t);
      records.push_back(msi);
    }
    return true;
  }
  return false;
}

bool CMarketStream::GetLastBondDeal(const std::vector<BrokerKey>& brokerKeys,
                                    std::vector<MarketStreamInfo>& records,
                                    time_t t, uint32_t day) {
  uint32_t UnDay = day <= 0 ? GetBondDealInvalidateDay() : day;
  records.reserve(brokerKeys.size());
  SCOPED_MUTEX(m_lock);
  RTDealList* rdlist = GetList(UnDay);
  if (!rdlist) return false;

  for (int i = 0; i < (int)brokerKeys.size(); ++i) {  // 取全部broker最新成交
    MarketStreamInfo msi;
    const BrokerKey& key = brokerKeys[i];
    char szBondKey[32] = {0};
    char szListedMarket[32] = {0};
    const CBondInfo& bi = CBondContainer::instance().ElementAtR(key.m_nIndex);
    FIELDCOPY(szBondKey, bi.GetBondKey());
    FIELDCOPY(szListedMarket, bi.GetListedMarket());
    if (CBrokerList::IsValidBroker(atoi(key.GetBrokerId().c_str()))) {
      rdlist->FindRecord(szBondKey, szListedMarket, key.GetBrokerId().c_str(),
                         msi, t);
      records.push_back(msi);
      continue;
    }

    MarketStreamInfo lastMsi;
    char sBroker[33] = {0};
    bool found = false;
    const CBrokerList::BrokerMap& brokerMap =
        SSUserAccountManager::instance().GetReadonlyBroker().GetBrokerMap();
    for (CBrokerList::BrokerMap::const_iterator it = brokerMap.begin();
         it != brokerMap.end(); ++it) {
      memset(sBroker, 0, sizeof(sBroker));
      FMTBUF(sBroker, "%d", it->second.m_brokerID);
      if (rdlist->FindRecord(szBondKey, szListedMarket, sBroker, msi, t)) {
        if (lastMsi.m_modify_time < msi.m_modify_time) lastMsi = msi;
      }
    }
    records.push_back(lastMsi);
  }

  return true;
}

void CMarketStream::InsertCashData(xQBMarketStreamList_c* pLstdata, bool& ret) {
  // 线程:后台线程ProcessQpidMessage
  // 作用:用来收取推动来的成交信息
  BASSERTS(IsObjectValid(), "");
  if (pLstdata && pLstdata->m_List.size() > 0) {
    // 初步过滤
    std::vector<ParsedStreamInfo> batch;
    if (FilterDealList(pLstdata, batch) && batch.size() > 0) {
      // log_warining("[STREAM][INFO]收到成交信息 %d 条,其中 %d 条无效",
      // pLstdata->m_List.size(), (pLstdata->m_List.size() - batch.size()));
      SCOPED_MUTEX(m_lock);
      if (!IsObjectValid()) return;
      int id = m_nextid++;
      for (int i = 0; i < (int)batch.size(); i++) {
        ParsedStreamInfo& si = batch[i];
        if (si.type == kStreamTypeBond)
          bondStream().GetOrCreateList(si.unDay)->InsertDirectly(si.info,
                                                                 si.unDay, id);
        else if (si.type == kStreamTypeExchange) {
          CMarketStream* pex = &exchangeStream();
          BASSERTS(pex, "");
          if (pex)
            pex->GetOrCreateList(si.unDay)->InsertDirectly(si.info, si.unDay,
                                                           id);
        }
      }
    }
    ret = batch.size() > 0 ? true : false;
  }
}

void CMarketStream::InsertCFETSCashData(
    xQBMarketStreamList_c* pLstdata, uint32_t DealType,
    bool& ret) {  // 线程:后台线程ProcessQpidMessage
  // 作用:用来收取推动来的成交信息
  if (pLstdata && pLstdata->m_List.size() > 0) {
    // 初步过滤
    std::vector<ParsedStreamInfo> batch;
    if (FilterCFETSDealList(pLstdata, batch, DealType) && batch.size() > 0) {
      log_debug("[STREAM][CFETS][INFO]收到成交信息 %d 条,其中 %d 条无效",
                pLstdata->m_List.size(),
                (pLstdata->m_List.size() - batch.size()));
      SCOPED_MUTEX(m_lock);
      if (!IsObjectValid()) return;
      int id = m_nextid++;
      for (int i = 0; i < (int)batch.size(); i++) {
        ParsedStreamInfo& si = batch[i];
        if (si.type == kStreamTypeCFets)
          GetOrCreateList(si.unDay)->InsertDirectly(si.info, si.unDay, id);
      }
    }
    ret = batch.size() > 0 ? true : false;
  }
}

bool CMarketStream::IsObjectValid() const { return m_tag == OBJECT_TAG; }
CMarketStream::RTDealList* CMarketStream::GetOrCreateList(DayType day) {
  SCOPED_MUTEX(m_lock);
  if (m_last && m_last->GetDay() == day) return m_last;

  DayDealMap::iterator it = m_deal_map.find(day);
  if (it != m_deal_map.end()) {
    m_last = it->second;
  } else {
    DayStateMap::iterator sit = m_deal_states.find(day);
    int64 state = (sit == m_deal_states.end()) ? 0 : sit->second;
    state++;
    m_deal_states[day] = state;
    m_last = new RTDealList(day, state);
    m_deal_map[day] = m_last;
  }

  return m_last;
}
CMarketStream::RTDealList* CMarketStream::GetList(DayType day) {
  if (day <= 0) day = GetBondDealInvalidateDay();
  DayDealMap::iterator it = m_deal_map.find(day);
  if (it != m_deal_map.end()) return it->second;
  return nullptr;
}
namespace {
bool IsCompanyIdPassed(const char* companyId, int corp) {
  int nTempCorp = 0;
  if (strcmp(companyId, "c") == 0)
    nTempCorp = 'c';
  else if (strcmp(companyId, "e") == 0)
    nTempCorp = 'e';
  else
    nTempCorp = atoi(companyId);

  if (corp == 0) {  // 成交数据 0为所有Broker
    if (!(nTempCorp > 0 && nTempCorp <= 6) || nTempCorp == 98)
      return false;
    else
      return true;
  }
  return corp == nTempCorp;
}
}  // namespace
bool CMarketStream::GetMarketStreamList(const MarkStreamListArg& arg,
                                        MSInfoList& mkList,
                                        MSRecMap* mkRecordMap) {
  /*
          用以替换CMarketStream中的以参数形式返回CMarketStreamlst的接口
          经过调查,绝大部分不需要CMarketStreamlst中的m_Record部分,为避免性能损失,将此拆成两部分,第二部分可选

          过滤总共分为四个步骤:
          1.按照日期过滤,重复日期自动忽略跳过
          2.按照MarketStreamInfo.m_company_id过滤,
                  如果arg.corp<0则直接跳过,如果arg.corp==0,检查m_company_id正确性
                  其他情况,只有当arg.corp==atoi(MarketStreamInfo.m_company_id)才通过
          3.按照债券有效性和MarktType过滤(隐藏的)
                  债券有效性是必须的,MarktType则不是必须的
                  如果arg.mark_type<0则对MarktType过滤直接通过,表示用户没有设置过
          4.按照emMktStreamPage类型过滤
                  这也是可选的,设置-1将直接忽略
  */
  int ncount = 0;
  const std::vector<uint32_t>& days = arg.days;
  std::vector<DealIndex> indexes;
  SCOPED_MUTEX(m_lock);
  for (int i = 0; i < (int)days.size(); i++) {
    if (i > 0 && IsExistInVec(days, i, days[i]))  // 日期去重
      continue;
    indexes.clear();
    RTDealList* rdlist = GetList(arg.days[i]);
    if (rdlist == nullptr) continue;
    IRecordMap* values = rdlist->GetKeys(indexes);
    for (int k = 0; k < (int)indexes.size(); k++) {
      const void* ptr = values->FindRecord(indexes[k]);
      if (ptr == nullptr) continue;
      const Record& rd = *((const Record*)ptr);
      if (arg.corp >= 0 && !IsCompanyIdPassed(rd.m_company_id, arg.corp))
        continue;

      int nIndex = CBondContainer::instance().GetBondIndex(
          rd.m_body.m_bondkey, rd.m_body.m_listedmarket);
      if (!CBondContainer::instance().IsValidIndex(nIndex)) continue;
      const CBondInfo& bond = CBondContainer::instance().ElementAtR(nIndex);
      if (arg.mark_type >= 0 && !IsBondPassType(bond, (MarktType)arg.mark_type))
        continue;
      if (arg.page_type >= 0 &&
          !IsBondPassPage(bond, (emMktStreamPage)arg.page_type))
        continue;
      ncount++;
      mkList.push_back(rd);
      if (mkRecordMap && arg.page_type != mktPage_NCD) {
        MarketStreamRec stat;
        std::string combBondKey = bond.GetCombBondKey();
        if (mkRecordMap->find(combBondKey) == mkRecordMap->end() &&
            rdlist->GetStat(combBondKey, stat)) {
          // 统计信息按照债券key来覆盖插入
          mkRecordMap->insert(MSRecMap::value_type(combBondKey, stat));
        }
      }
    }
  }
  return ncount > 0;
}

bool CMarketStream::GetTopRecords(const MarketStreamTopArg& arg,
                                  CMarketToprec& MarketInfo) {
  // 用以替换CMarketStream中返回CMarketToprec的接口
  // 用途:获取选定日期和指定类型的所有成交信息中,最活跃的前指定个数的成交信息
  const std::vector<uint32_t>& days = arg.days;
  CMarketToprec RecordInfo;
  SCOPED_MUTEX(m_lock);
  if (arg.mark_type < 0) {
    detail::RealTimeDealStat statmap;
    for (int i = 0; i < (int)days.size(); i++) {
      if (i > 0 && IsExistInVec(days, i, days[i])) continue;
      RTDealList* rdlist = GetList(arg.days[i]);
      if (rdlist) statmap.AddStatMap(rdlist->GetStatObject());
    }
    return detail::MSUtil::SortMarketRecord(statmap.GetStats(), RecordInfo,
                                            MarketInfo, arg.count);
  } else {
    std::map<std::string, MarketStreamRec> mapRecordTp;
    for (int i = 0; i < (int)days.size(); i++) {
      if (i > 0 && IsExistInVec(days, i, days[i])) continue;
      MSInfoList mkList;
      MSRecMap mkMap;
      MarkStreamListArg mkArg;
      mkArg.days.push_back(days[i]);
      mkArg.mark_type = arg.mark_type;
      GetMarketStreamList(mkArg, mkList, &mkMap);
      if (mkMap.size() > 0) {
        MarketInfo.SetSize(mkList.size() + MarketInfo.m_nTotal);
        for (std::map<std::string, MarketStreamRec>::iterator itrRe =
                 mkMap.begin();
             itrRe != mkMap.end(); itrRe++) {
          detail::MSUtil::InsertMarketRecord(itrRe->first.c_str(),
                                             itrRe->second, mapRecordTp);
        }
      }
    }
    if (mapRecordTp.size() < 0) return false;
    return detail::MSUtil::SortMarketRecord(mapRecordTp, RecordInfo, MarketInfo,
                                            arg.count);
  }
}
bool CMarketStream::GetMarketInfo(const MarketStreamInfoArg& arg,
                                  CMarketInfoMap& infoMap) {
  // 用以替换CMarketStream中的返回CMarketInfoMap的接口
  // 用途:根据各种条件,对选定日期的成交信息进行计数统计
  BASSERTS(arg.stat_type > kStatTypeMin && arg.stat_type < kStatTypeMax, "");
  SCOPED_MUTEX(m_lock);
  int ncount = 0;
  const std::vector<uint32_t>& days = arg.days;
  std::vector<DealIndex> indexes;
  for (int i = 0; i < (int)days.size(); i++) {
    if (i > 0 && IsExistInVec(days, i, days[i])) continue;
    RTDealList* rdlist = GetList(arg.days[i]);
    if (rdlist == nullptr) continue;
    IRecordMap* values = rdlist->GetKeys(indexes);
    for (int k = 0; k < (int)indexes.size(); k++) {
      const void* ptr = values->FindRecord(indexes[k]);
      if (ptr == nullptr) continue;
      const Record& rd = *((const Record*)ptr);
      if (UpdateMarketInfoMap(rd, (MarktType)arg.mark_type, arg.stat_type,
                              infoMap))
        ncount++;
    }
  }
  infoMap.SetSize(ncount);
  return ncount > 0;
}

//////////////////////////////////////////////////////////////////////////

const MarketStreamInfo& CMarketStream::iterator::operator*() const {
  return m_enumerator->GetAt(m_index);
}
const MarketStreamInfo* CMarketStream::iterator::operator->() const {
  return &(m_enumerator->GetAt(m_index));
}
CMarketStream::Enumerator::Enumerator(CMarketStream* object, uint32_t day,
                                      bool all)
    : /*m_scoped(object->m_lock),*/ m_values(nullptr), m_stats(nullptr) {
  RTDealList* dealList = object->GetList(day);
  if (dealList) {
    m_values = dealList->GetKeys(m_indexes, all);
    m_stats = dealList->GetStatObject();
  }
}

CMarketStream::iterator CMarketStream::Enumerator::begin() const {
  return CMarketStream::iterator(0, this);
}
bool CMarketStream::Enumerator::findrec(const std::string& combBondKey,
                                        MarketStreamRec& rec) const {
  if (m_stats && m_stats->GetStat(combBondKey, rec)) return true;
  return false;
}

const MarketStreamInfo& CMarketStream::Enumerator::GetAt(int nIndex) const {
  // 在调用此接口之前,必须通过iterator it!=end();
  BASSERTS(m_values, "m_values 必须不为0,m_indexes.size()=%d",
           m_indexes.size());
  BASSERTS(nIndex >= 0 && nIndex < (int)m_indexes.size(),
           "nIndex:%d m_indexes.size()=%d", nIndex, m_indexes.size());
  return *((const MarketStreamInfo*)m_values->FindRecord(m_indexes[nIndex]));
}
